// @Author EthanScriptOn
// @Desc
package mysql_operator

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_expansion_package/connect"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator/op_example"
	"time"
)

type MysqlQueryOperator struct {
	*op_example.LoopDetectionMultiChannelDataOperator
	*identity.IdentityInformationCarrier
}

func (m *MysqlQueryOperator) ExecuteWithLoop(ctx *actuator.SrfSwitchBridgeActuatorContext, operatorConfigIdentifier interface{}, structureParam interface{}) (err error) {
	commandSql, ok := structureParam.(string)
	if !ok {
		err = fmt.Errorf("structureParam is not string")
		_ = ctx.Record(actuator.OperatorExecute, err.Error())
		return err
	}
	if commandSql == "" {
		err = fmt.Errorf("commandSql is empty")
		_ = ctx.Record(actuator.OperatorExecute, err.Error())
		return err
	}
	operatorIdentifier, err := m.IdentityInformationCarrier.GenerateUniqueIdentifier()
	if err != nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error generating unique identifier: %v", err))
		return err
	}
	pipeline := m.GetDataPipeline(operatorIdentifier)
	if pipeline == nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error getting data pipeline: %v", err))
		return err
	}

	config := operator.GetOperatorConfig(operatorConfigIdentifier)
	if config == nil {
		return fmt.Errorf("config is nil")
	}
	operatorConfig, ok := config.(*extend_config.MysqlQueryOperatorConfig)
	if !ok {
		return fmt.Errorf("config is not *MysqlQueryOperatorConfig")
	}

	db, err := connect.GetMysqlQueryDB()
	if err != nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error getting mysql query: %v", err))
		return err
	}
	defer db.Close()

	_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("mysql query: %s", commandSql))
	ticker := time.NewTicker(operatorConfig.DelayTime)
	for {
		select {
		case <-ticker.C:
			rows, err := db.Query(commandSql)
			if err != nil {
				err = fmt.Errorf("failed to execute SQL query: %v", err)
				return err
			}
			pipeline <- rows
		}
	}
}
